/** https://github.com/apache/rocketmq-clients/blob/golang/v5.1.3/golang/example/consumer/simple_consumer/main.go */
package main

import (
	"fmt"
	"gitee.com/zackeus/go-zero/core/logx"
	"os"
	"time"
)

import (
	"context"
	"github.com/apache/rocketmq-clients/golang/v5"
	"github.com/apache/rocketmq-clients/golang/v5/credentials"
)

const (
	Topic     = "test-topic"
	GroupName = "test-group"
	// Endpoint 多个封号分割
	Endpoint  = "192.168.137.31:8081"
	Region    = "A"
	AccessKey = "rmq-admin"
	SecretKey = "baicaf5678#"
)

var (
	// maximum waiting time for receive func
	awaitDuration = time.Second * 5
	// 一次接收的最大消息数
	maxMessageNum int32 = 1
	// invisibleDuration 制消息的「不可见时长」 should > 20s
	// 在这段时间内，同消费组的其他消费者无法获取到这些消息，确保当前消费者有足够时间处理消息
	invisibleDuration = time.Second * 20
)

func main() {
	/* 日志输出至控制台 */
	os.Setenv(golang.ENABLE_CONSOLE_APPENDER, "true")
	os.Setenv(golang.CLIENT_LOG_LEVEL, "error")
	golang.ResetLogger()
	// new simpleConsumer instance
	simpleConsumer, err := golang.NewSimpleConsumer(&golang.Config{
		Endpoint:      Endpoint,
		ConsumerGroup: GroupName,
		NameSpace:     Region,
		Credentials: &credentials.SessionCredentials{
			AccessKey:    AccessKey,
			AccessSecret: SecretKey,
		},
	},
		golang.WithClientFuncForSimpleConsumer(golang.NewClient), //构建client函数
		golang.WithSimpleAwaitDuration(awaitDuration),
		golang.WithSimpleSubscriptionExpressions(map[string]*golang.FilterExpression{
			Topic: golang.SUB_ALL,
		}),
	)
	if err != nil {
		logx.Error(err)
		return
	}
	// start simpleConsumer
	err = simpleConsumer.Start()
	if err != nil {
		logx.Error(err)
		return
	}

	sig := make(chan os.Signal)
	ctx, cancel := context.WithCancel(context.Background())
	go func() {
		for {
			select {
			case <-ctx.Done():
				// 收到退出信号，停止循环
				fmt.Println("receive loop is stopping...")
				return
			default:
				// 正常接收消息
				fmt.Println("start receive message")
				mvs, err := simpleConsumer.Receive(ctx, maxMessageNum, invisibleDuration)
				if err != nil {
					fmt.Println(err)
				}
				// ack message
				for _, mv := range mvs {
					simpleConsumer.Ack(ctx, mv)
					fmt.Println(mv)
				}
			}
		}
	}()

	<-sig
	cancel()
	if err := simpleConsumer.GracefulStop(); err != nil {
		fmt.Printf("shutdown Consumer error: %s", err.Error())
	}
}
